#!/bin/bash
#设置任务异常退出
set -eo pipefail
dt='{{ execution_date | cst_ds }}'
name='网点未规范登记退回件表'
# 在退出时清理文件
trap "rm -f /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv && echo 'removed tmp files'" EXIT
# 创建txt文件
touch /tmp/${name}_${dt}.txt
#将中文字段名输入文件顶部
#将查询结果追加到文件中
{
  echo -e "签收日期\t运单号\t揽收网点\t始发集散点\t始发转运中心\t问题件登记网点\t问题件二级类型\t签收网点名称\t原派件网点\t原派件网点是否是异常网点\t原运单三段码是否是异常段码\t异常第三段码\t">/tmp/${name}_${dt}.txt;
  spark-sql --driver-memory 8G --num-executors 4 --executor-memory 20g --executor-cores 4  --conf spark.executor.memoryOverhead=4g --conf spark.driver.maxResultSize=3g -S -e "
    --中心发件信息
--运单表
with tmp_waybill as (
  select
     waybill_no         --运单号
    ,sign_network_code  --签收网点
    ,sign_network_name
    ,pick_network_code  --揽收网点
    ,pick_network_name
    ,sign_time          --签收时间
    ,first_code   --一段码
    ,second_code  --二段码
    ,third_code   --三段码
  from jms_dwd.dwd_yl_oms_oms_waybill_incre_dt t1
  where dt between date_sub('{{ execution_date | cst_ds }}',30) and '{{ execution_date | cst_ds }}'
    and date(sign_time) = '{{ execution_date | cst_ds }}'
)

--退转件管理表
, tmp_reback as (
  select
    waybill_no
  from jms_dwd.dwd_tab_reback_transfer_express_base
  where date(apply_time) >= date_sub('{{ execution_date | cst_ds }}',61)
  group by waybill_no
)

--拦截件配置表
, tmp_interceptor as (
  select
     data_type    --'网点，异常区域，异常三段码',
    ,network_code  --网点
    ,network_name
    ,the_third_code  --第三段码
  from jms_dim.dim_yl_oms_order_interceptor_base
  where IS_ENABLE = 1
    and IS_DELETE = 0
)

--未签收概要宽表：取运单的始发集散、始发中心
, tmp_unsign as (
  select
    waybill_no,
    max(first_nodal_network_code ) as first_nodal_network_code   , --始发集散
    max(first_center_network_code) as first_center_network_code   --始发转运中心
  from jms_dwd.dwd_wide_summary_waybill
  where dt between '2022-10-15' and '{{ execution_date | cst_ds }}'
  group by waybill_no
)

--问题件表：取30天数据，根据运单取最新
, tmp_difficute as (
  select * from (
    select
       waybill_no
      ,scan_site
      ,scan_site_code   --问题件登记网点
      ,problem_type as difficute_type --     --问题件二级类型
      ,scan_time
      ,row_number() over(partition by waybill_no order by scan_time desc) rn
    from jms_dwd.dwd_tab_barscan_difficult_base_dt
    where dt between date_sub('{{ execution_date | cst_ds }}',30) and date_sub('{{ execution_date | cst_ds }}',0)
  ) tmp where rn = 1
)

--一段码维护表
, tmp_first_code as (
  select * from (
    select
       code
      ,center_code
      ,create_time
      ,update_time
      ,effect_time
      ,invalid_time
      ,row_number() over(partition by code order by update_time desc) rn
    from jms_dim.dim_mysql_lmdm_sys_first_code
    where is_delete = 1
      and is_enable = 1
  ) tmp where rn = 1
)

--二段码维护表
, tmp_second_code as (
  select
  code,
  network_code,
  center_code,
  network_name
  from jms_dim.dim_mysql_lmdm_sys_second_code
  where is_delete = 1
    and is_enable = 1
)

select
  date(t1.sign_time) as sign_date    --签收日期
 ,t1.waybill_no                      --运单号
 ,t1.pick_network_name
 ,t10.name as first_nodal_network        --始发集散点
 ,t11.name as first_center_network       --始发转运中心
 ,t5.scan_site  --问题件登记网点
 ,t5.difficute_type  --问题件二级类型
 ,t1.sign_network_name   --签收网点名称
 ,t7.network_name   --原派件网点
 ,case when t8.network_code is not null then '是' else '否' end as old_sign_network_is_exception --原派件网点是否是异常网点
 ,case when t9.the_third_code is not null then '是' else '否' end as old_third_code_is_exception --原派件网点是否是异常网点
 ,t9.the_third_code    --异常第三段码
 --,date(t1.sign_time) as dt
from tmp_waybill t1
left join tmp_difficute t2   on t1.waybill_no = t2.waybill_no
left join tmp_reback t3      on t1.waybill_no = t3.waybill_no
left join tmp_unsign t4      on t1.waybill_no = t4.waybill_no
left join tmp_difficute t5   on t1.waybill_no = t5.waybill_no
left join tmp_first_code t6  on t1.first_code = t6.code
left join tmp_second_code t7 on t1.second_code = t7.code and t6.center_code = t7.center_code
left join (select network_code from tmp_interceptor where data_type = 'network' and network_code is not null group by network_code) t8 on t7.network_code = t8.network_code
left join (select the_third_code from tmp_interceptor where data_type = 'the_third_segment_code' and the_third_code is not null group by the_third_code) t9 on t1.third_code = t9.the_third_code
left join jms_dim.dim_network_whole_massage t10 on t4.first_nodal_network_code = t10.code
left join jms_dim.dim_network_whole_massage t11 on t4.first_center_network_code = t11.code
where t1.pick_network_code = t1.sign_network_code
  and t2.waybill_no is not null --有问题件登记记录
  and t3.waybill_no is null  --无退件记录
; " >>/tmp/${name}_${dt}.txt
}

# 替换分隔符，并输出csv
cat /tmp/${name}_${dt}.txt | tr -s '\t' ',' > /tmp/${name}_${dt}.csv
# 压缩并删除 csv 文件
zip -jrm /tmp/${name}_${dt}.zip /tmp/${name}_${dt}.csv
# 上传文件
/usr/local/python3/bin/python3  /usr/local/airflow/dags/utils/operators/upload_oss.py ${name}/${name}_${dt}.zip /tmp/${name}_${dt}.zip
# zip 文件将在进程退出后被自动清理

#声明文件大小
data_size=$(ls -lh /tmp/${name}_${dt}.zip | awk '{print $5}')
#打印文件信息
Message="${name}_${dt}.zip 上传成功 大小: $data_size"
#配置钉钉告警
curl 'https://oapi.dingtalk.com/robot/send?access_token=816a8f35fd84432ffa6bff9d7714abc322628bb4a8ded86908550d2335aba613' \
   -H 'Content-Type: application/json' \
   -d "
  {\"msgtype\": \"text\",
    \"text\": {
        \"content\": \"监控内容:$Message\"
     }
  }" || true
rm -f  /tmp/${name}_${dt}.txt